package com.ztesoft.web.cleanup;

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;

import javax.sql.DataSource;

import org.springframework.jdbc.datasource.DataSourceUtils;
import org.springframework.jdbc.support.JdbcUtils;

import com.ztesoft.core.db.ds.DynamicDataSourceManager;
import com.ztesoft.core.spring.context.SpringApplicationContextHolder;
import com.ztesoft.framework.exception.BaseAppException;
import com.ztesoft.framework.log.ZTEsoftLogManager;
import com.ztesoft.framework.util.DateUtils;
import com.ztesoft.framework.util.Utils;
import com.ztesoft.web.baseconfig.db.dao.DmSrcTableDao;
import com.ztesoft.web.baseconfig.db.po.DmPlanPO;
import com.ztesoft.web.baseconfig.db.po.DmSrcTablePO;
import com.ztesoft.web.baseconfig.db.po.DmStrategyPO;
import com.ztesoft.web.cleanup.service.CleanUpTaskService;
import com.ztesoft.web.common.customdb.dao.DmSrcTableCustomDao;
import com.ztesoft.web.common.customdb.dao.DmTaskStageCustomDao;
import com.ztesoft.web.common.customdb.dao.dto.DmTaskStageDto;
import com.ztesoft.web.common.db.arg.DmArg;
import com.ztesoft.web.common.db.dao.DmDao;
import com.ztesoft.web.common.db.dao.DmSecondaryStageDao;
import com.ztesoft.web.common.db.po.DmPO;
import com.ztesoft.web.common.db.po.DmSecondaryStagePO;
import com.ztesoft.web.resourcecheck.ResourceCheckJob;
import com.ztesoft.web.syncbaseinfo.service.BaseInfoService;
import com.ztesoft.web.syncbaseinfo.service.impl.BaseInfoServiceImpl;
import com.ztesoft.web.taskgenerator.db.dao.DmTaskDao;
import com.ztesoft.web.taskgenerator.db.po.DmTaskPO;

/**
 * 清理任务处理 
 * 
 * @author lin.weibin@zte.com.cn
 * @date 2016年1月13日
 * @since 1.0
 */
public class CleanUpTask implements Callable<DmTaskStageDto> {
	
	private static final ZTEsoftLogManager LOGGER = ZTEsoftLogManager.getLogger(CleanUpTask.class);
	
	private DmSrcTableCustomDao srcTableCustomDao;

	private DmSecondaryStageDao secondaryStageDao;
	
	private DmDao dmDao;
	
	private DmTaskStageDto taskStagePO;
	
	private DmPlanPO planPO;
	
	private DmSrcTablePO srcTablePO;
	
	private CleanUpTaskService cleanUpTaskService;

	private DmTaskStageCustomDao taskStageCustomDao;
	
    private ResourceCheckJob resourceCheckJob;
    
	private BaseInfoService baseInfoService;
	
	private String[] dbConfig;
	
	private DmStrategyPO stgPO;
	
	private Integer stage;
	
	private DmTaskDao dmTaskDao;
	
	private DynamicDataSourceManager dynamicDataSourceManager;
	
	private DmSrcTableDao dmSrcTableDao;
	
	public CleanUpTask(DmTaskStageDto taskStagePO, DmPlanPO planPO, DmSrcTablePO srcTablePO,String[] dbConfig, DmStrategyPO stgPO, Integer stage) {
		LOGGER.info("{0} initialize cleanup task class", Thread.currentThread());
		this.taskStagePO = taskStagePO;
		this.planPO = planPO;
		this.srcTablePO = srcTablePO;
		this.dbConfig = dbConfig;
		this.stgPO = stgPO;
		this.stage = stage;
		resourceCheckJob = SpringApplicationContextHolder.getBean(ResourceCheckJob.class);
		dmSrcTableDao = SpringApplicationContextHolder.getBean(DmSrcTableDao.class);
		srcTableCustomDao = SpringApplicationContextHolder.getBean(DmSrcTableCustomDao.class);
		dmDao = SpringApplicationContextHolder.getBean(DmDao.class);
		cleanUpTaskService = SpringApplicationContextHolder.getBean(CleanUpTaskService.class);
		taskStageCustomDao = SpringApplicationContextHolder.getBean(DmTaskStageCustomDao.class);
		secondaryStageDao = SpringApplicationContextHolder.getBean(DmSecondaryStageDao.class);
		baseInfoService = SpringApplicationContextHolder.getBean(BaseInfoServiceImpl.class);
		dmTaskDao = SpringApplicationContextHolder.getBean(DmTaskDao.class);
		dynamicDataSourceManager = (DynamicDataSourceManager) SpringApplicationContextHolder.getBean("dynamicDataSourceManager");
	}
	

	@Override
	public DmTaskStageDto call() {
		List<DmSecondaryStagePO> secondaryStagePOList = null;
		List<DmSrcTablePO> refList = null;
		String checkState = null;
		String dmTableName = "DM_"+taskStagePO.getCheckObjName();;
		java.util.Date now = DateUtils.getDBCurrentTime();
		int dloop = 1;
		int count = 0;
		int record = 0;
		int resultCount = 0;
		LOGGER.info("{0} 1、更新环节记录--",Thread.currentThread());
		taskStagePO.setState("00R");
		taskStagePO.setStateDate(now);
		taskStagePO.setTaskStartTime((taskStagePO.getTaskStartTime()!=null)?null:now);
		taskStagePO.setTaskEndTime(null);
		taskStageCustomDao.updateTaskStageState(taskStagePO);

		if (planPO.getOperType() == 0) {
			secondaryStagePOList = secondaryStageDao.selectByInfoId(taskStagePO.getDmInfoId());
			LOGGER.info(Thread.currentThread()+"2、从表环节表："+secondaryStagePOList.size());
//			for (DmSecondaryStagePO secondaryStagePO : secondaryStagePOList) {//如果有环节记录改状态为执行中		
//				secondaryStagePO.setTaskStartTime(now);
//				secondaryStagePO.setStateDate(now);
//				secondaryStagePO.setState("00R");
//				dmSecondaryStageDao.updateByPrimaryKeySelective(secondaryStagePO);
//			}
			
		} else if (planPO.getOperType() == 1) {
			refList = srcTableCustomDao.selectByStrategyRef(taskStagePO.getDmStrategyId());
		}
		while(true) {
//			if (stage == 4) {//
//				stage =null;
//				dmDao.updateByLimit(dmTableName, "00C", DateUtils.getDBCurrentTime(), taskStagePO.getDmTaskId(), 
//						taskStagePO.getDmStrategyId(), "0CF", stgPO.getBatchCommitCount(), null);
//			}
//			dloop++;
//			try {
//				//开发测试~
//				Thread.sleep(30*1000);
//			} catch (InterruptedException e1) {
//			}
			long tt = System.currentTimeMillis();
			LOGGER.info(Thread.currentThread()+"3、使用任务ID和策略ID按批次(每批次数量默认1000条)提取任务明细记录--"+ stgPO.getBatchCommitCount());
			List<DmPO> dmPOList = dmDao.selectByLimit(dmTableName, taskStagePO.getDmTaskId(), 
					taskStagePO.getDmStrategyId(), "00C", stgPO.getBatchCommitCount());
			LOGGER.info(Thread.currentThread()+"3-1、取待清理数据::--"+ dmPOList.size());
			if (Utils.isEmpty(dmPOList)) {//若无数据，则退出改环节记录状态为00A state_date end_time
				now = DateUtils.getDBCurrentTime();
				taskStagePO.setStateDate(now);
				taskStagePO.setTaskEndTime(now);
				taskStagePO.setTaskStartTime(null);
				taskStagePO.setState("00A");
				taskStagePO.setCheckState("success");
				//
				if (stage == null) {//正常流程走这个~ add by lin.weibin 2015.12.15					
					if (planPO.getTableAnalyseFlag() == 1 || planPO.getIndexAnalyseFlag() == 1) {//
						DmTaskPO taskPO = dmTaskDao.selectByPrimaryKey(taskStagePO.getDmTaskId());
						if ("F".equals(taskPO.getNeedSubtask())) {//不需要子任务，则启动表/索引分析线程
							new AnalyzeThread(taskStagePO.getCheckObjName(), srcTablePO.getDbLinkId(), srcTablePO.getDbUserName(), dbConfig)
							.start();
							DmSrcTablePO refSrcTablePO;//从表
							String[] dbConfigRef = null;
							secondaryStagePOList = secondaryStageDao.selectByInfoId(taskStagePO.getDmInfoId());
							for (DmSecondaryStagePO secondaryStagePO : secondaryStagePOList) {
								refSrcTablePO = dmSrcTableDao.selectByPrimaryKey(secondaryStagePO.getSourceDataId());
								try {
									dbConfigRef = baseInfoService.getDbConfig(refSrcTablePO.getDbLinkId(), refSrcTablePO.getDbUserName());
								} catch (BaseAppException e) {
								}
								new AnalyzeThread(secondaryStagePO.getCheckObjName(), refSrcTablePO.getDbLinkId(), refSrcTablePO.getDbUserName(), dbConfigRef)
								.start();
							}
						}
					}
				}
				taskStagePO.setCheckCount(resultCount);
				return taskStagePO;
			}
			 try {
				//执行删除并更新任务环节表
				record = cleanUpTaskService.addCleanUp(stage, dmPOList, secondaryStagePOList, refList, srcTablePO, taskStagePO, dloop++, dbConfig);
				resultCount += record;
				taskStagePO.setCheckCount(record);
				//更新任务明细状态
			
				dmDao.updateByLimit(dmTableName, "00A", DateUtils.getDBCurrentTime(), taskStagePO.getDmTaskId(), 
						taskStagePO.getDmStrategyId(), "00C", stgPO.getBatchCommitCount(), null);
					
				if (stage == null) {//备份取消调用直接删除，不移动数据~	
					taskStagePO.setCheckDataSize(BigDecimal.valueOf(
							((double)(srcTablePO.getOneRecordSize()*record))/1024)
							.setScale(2, RoundingMode.HALF_UP));
					taskStagePO.setStateDate(DateUtils.getDBCurrentTime());
					taskStageCustomDao.updateTaskStageCheckCount(taskStagePO);
					
					// 清理结束历史表归档
					LOGGER.info(Thread.currentThread()+"批量插入DM历史表 ..."+(dmTableName+"_his::"));
					count = dmTaskDao.selectExistTableName(dmTableName+ "_his");
					if(count==0){
						LOGGER.info(Thread.currentThread()+"创建一张dm_tableName表"+(dmTableName+ "_his"));
						dmTaskDao.addCreateDmTable(dmTableName+ "_his", "DM");
					}
					List<DmPO> dmList = dmDao.selectByLimit(dmTableName, taskStagePO.getDmTaskId(), 
							taskStagePO.getDmStrategyId(), "00A", null);
					dmDao.insertBatch(dmTableName+"_his", dmList);
				}
				LOGGER.info(Thread.currentThread()+"删除明细数据");
				DmArg arg = new DmArg();
			    DmArg.DmCriteria criteria = arg.createCriteria();
			    criteria.andDmTaskIdEqualTo(taskStagePO.getDmTaskId());
			    criteria.andDmStrategyIdEqualTo(taskStagePO.getDmStrategyId());
			    criteria.andStateEqualTo("00A");
			    dmDao.deleteByArg(dmTableName, arg);
				LOGGER.debug("|||第 {0} 批次 use:"+(System.currentTimeMillis()-tt)+"ms|策略{1}| to insert into his'table",dloop-1, taskStagePO.getDmStrategyId());
			} catch (BaseAppException e) {//若失败，则退出改主、从环节记录状态为00F state_date end_time dm 状态为0CF 
				LOGGER.error("清理失败cleanup error:",e);
				now = DateUtils.getDBCurrentTime();
				dmDao.updateByLimit(dmTableName, "0CF", DateUtils.getDBCurrentTime(), taskStagePO.getDmTaskId(), 
						taskStagePO.getDmStrategyId(), "00C", stgPO.getBatchCommitCount(), e.getMessage());
//				taskStagePO.setCheckCount(record);
//				taskStagePO.setCheckDataSize(BigDecimal.valueOf(srcTablePO.getOneRecordSize()*(record)));
				taskStagePO.setStateDate(now);
				taskStagePO.setTaskEndTime(now);
				taskStagePO.setState("00F");
				taskStagePO.setTaskStartTime(null);
				taskStagePO.setErrorDesc(e.getCode());
				taskStagePO.setCheckState("fail");
				LOGGER.error("清理失败托送告警信息-----");
				resourceCheckJob.pushMsg2IOCP("[cleanup error]strategyID:"+taskStagePO.getDmStrategyId(), 
						e.getCode(), e.getCode());
				taskStagePO.setCheckCount(resultCount);
				return taskStagePO;
			}
			 /*
			  * 暂停、取消判断
			  */
			 if (null == stage) {//			 
				 checkState = baseInfoService.doCheck(taskStagePO, planPO);
				 if (checkState != null) {
					taskStagePO.setCheckState(checkState);
					LOGGER.info("{0},暂停、取消处理{1}",Thread.currentThread(),checkState);
					taskStagePO.setCheckCount(resultCount);
					return taskStagePO;
				 }
			 }
		}
	}

	/**
	 * 表分析、索引分析
	 * @param tableName	表名
	 * @param dbLinkId	链接标志
	 * @param dbUserName 用户名
	 * @param dbConf 配置信息
	 */
	private void analyze(String tableName, Integer dbLinkId, String dbUserName, String[] dbConf) {
		LOGGER.info(Thread.currentThread()+">>>>{0}执行{1}表分析索引分析<<<<--begin...",dbConf[0],tableName);
		long tt = System.currentTimeMillis();
		DataSource dataSource = null;
		Connection conn = null;
		DatabaseMetaData metaData = null;
		PreparedStatement ps = null;
		ResultSet rs = null;
		ResultSet columns = null;
		List<String> pkList = null;
		List<String> colList = null;
		StringBuffer sqlStr = new StringBuffer();
		StringBuffer indexStr = new StringBuffer();
		try {
			pkList = new ArrayList<String>(2);
			colList = new ArrayList<String>(2);
			dataSource = dynamicDataSourceManager.getDruidDataSource(dbLinkId+dbUserName, dbConf);
			LOGGER.info(Thread.currentThread()+"获取链接信息中...");
			conn = DataSourceUtils.getConnection(dataSource);
			LOGGER.info(Thread.currentThread()+"获取链接信息ok...");
			metaData = conn.getMetaData();
			//主键字段
			columns = metaData.getPrimaryKeys(null, null, tableName);
			while (columns.next()) {
				pkList.add(columns.getString("COLUMN_NAME").toLowerCase());
			}
			//所有字段
			columns = metaData.getColumns(null, null, taskStagePO.getCheckObjName(),null);
			while (columns.next()) {
				colList.add(columns.getString("COLUMN_NAME").toLowerCase());
			}
//			System.err.println(pkList+"---\r\n"+colList);

			if ("oracle".equals(dbConf[0])) {
				if (planPO.getTableAnalyseFlag().intValue() == 1) {					
					sqlStr.append("analyze table ").append(tableName).append(" compute statistics;");
				}
				if (planPO.getIndexAnalyseFlag().intValue() == 1) {
					ps = conn.prepareStatement(SELECT_INDEX);
					ps.setString(1, tableName.toUpperCase());
					rs = ps.executeQuery();
					while (rs.next()) {
						indexStr.append("analyze index ").append(rs.getString(1)).append(" compute statistics;");
					}
					// close 
					JdbcUtils.closeStatement(ps);
					JdbcUtils.closeResultSet(rs);
					//
					if (indexStr.length()!=0) {						
						sqlStr = sqlStr.append(indexStr.toString());
					}
				}
			} else if ("mysql".equals(dbConf[0]) && (planPO.getTableAnalyseFlag() == 1 || planPO.getIndexAnalyseFlag() == 1)) {
				
				sqlStr.append("analyze table ").append(tableName).append(";");
				
			} else if ("informix".equals(dbConf[0]) && (planPO.getTableAnalyseFlag() == 1 || planPO.getIndexAnalyseFlag() == 1)) {
				sqlStr.append(INFORMIX_ANALYZE_TABLE).append(tableName).append(";");//表
				for (String pk : pkList) {//主键字段
					sqlStr.append(INFORMIX_ANALYZE_TABLE).append(tableName).append("(").append(pk).append(");");
				}
				for (String col : colList) {//
					if (!pkList.contains(col)) {
						sqlStr.append(INFORMIX_ANALYZE_MEDIUM_).append(tableName).append("(").append(col).append(");");
					}
				}
			}
			LOGGER.info("{0}analyze sql:{1}",Thread.currentThread(),sqlStr);
			if (sqlStr != null && Utils.notEmpty(sqlStr.toString())) {//
				if ("oracle".equals(dbConf[0])) {
					String[] sqlArr = sqlStr.toString().split(";");
					for (String sql : sqlArr) {						
						ps = conn.prepareStatement(sql);
						ps.execute();
					}
				} else {					
					ps = conn.prepareStatement(sqlStr.toString());
					ps.execute();
				}
			}
		} catch (SQLException e) {
			LOGGER.error("analyze err", e);
		} finally {
			JdbcUtils.closeStatement(ps);
			JdbcUtils.closeResultSet(rs);
			DataSourceUtils.releaseConnection(conn, dataSource);
			LOGGER.info(Thread.currentThread()+">>>>{0}执行{2}表分析索引分析<<<<--end::耗时{1}ms...",dbConf[0],(System.currentTimeMillis()-tt), tableName);
		}
	}
	
	/**
	 * 内部线程类，处理表分析索引分析~
	 * 
	 * 
	 * @author lin.weibin@zte.com.cn
	 * @date 2016年1月13日
	 * @since 1.0
	 */
	class AnalyzeThread extends Thread {
		String tableName;
		Integer dbLinkId;
		String dbUserName;
		String[] dbConf;
		
		public AnalyzeThread(String tableName, Integer dbLinkId, String dbUserName, String[] dbConf) {
			this.tableName = tableName;
			this.dbLinkId = dbLinkId;
			this.dbUserName = dbUserName;
			this.dbConf = dbConf;
		}

		public void run() {
			analyze(tableName, dbLinkId, dbUserName, dbConf);
		}
		
	}
	private static final String INFORMIX_ANALYZE_TABLE = "UPDATE STATISTICS high FOR TABLE ";
	private static final String INFORMIX_ANALYZE_MEDIUM_ =  "UPDATE STATISTICS medium FOR TABLE ";
	private static String SELECT_INDEX = " SELECT INDEX_NAME FROM USER_INDEXES WHERE TABLE_NAME = ? AND TABLE_TYPE = 'TABLE'";
//	SELECT INDEX_NAME FROM USER_INDEXES WHERE TABLE_NAME = ? AND TABLE_OWNER = ? AND TABLE_TYPE = 'TABLE'
}
